package spring.cloud.tasks.executor.executor;

import com.google.common.base.Joiner;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.common.HostUtil;
import spring.cloud.tasks.common.Symbols;
import spring.cloud.tasks.common.curator.Curator;
import spring.cloud.tasks.common.nodepaths.ExecutorNodePaths;

import java.util.ArrayList;
import java.util.List;

import static spring.cloud.tasks.common.Threads.sleep;


@Slf4j
@Setter
@Getter

public class ExecutorInitAndTaskInitService {
    private static final int WAIT_FOR_IP_NODE_DISAPPEAR_COUNT = 300;//300/60 如果五分钟ip瞬时节点还没有小时  那么代表确实存在相同执行器节点存在。
    private String executorId;
    private Curator curator;
    private String executorNodePath;
    private String executorSystemTimePath;
    private String executorIpNodePath;
    private String executorTaskIdListNodePath;
    //
    private TaskExecutorInitAndTaskItemAssignListenerService taskInitService;
    //##<!-我是优美的功能模块分割线->##

    public ExecutorInitAndTaskInitService(@NonNull String executorId, Curator curator) {
        this.executorId = executorId;
        this.curator = curator;

    }

    /**
     * 注册Executor
     */
    public void registerExecutor() throws Exception {
        checkExecutorNodeExists();
        registerExecutor0();
    }

    private void checkExecutorNodeExists() throws Exception {
        // 启动时检查本机与注册中心的时间误差秒数是否在允许范围
        String executorNodePath = ExecutorNodePaths.ROOT + "/" + executorId;
        String executorSystemTimePath = executorNodePath + "/systemTime";
        String executorIpNodePath = executorNodePath + "/ip";
        String executorTaskIdListNodePath = executorNodePath + "/taskIdList";
        this.executorNodePath = executorNodePath;
        this.executorSystemTimePath = executorSystemTimePath;
        this.executorIpNodePath = executorIpNodePath;
        this.executorTaskIdListNodePath = executorTaskIdListNodePath;
        //
        try {
            if (curator.checkExists(executorSystemTimePath)) {
                curator.delete(executorSystemTimePath);
            }
            curator.merge(executorSystemTimePath, "");
            /**
             * 启动前先检查本机与注册中心的时间误差秒数是否在允许范围和Executor是否已启用
             */
            long timeDiff = Math.abs(System.currentTimeMillis() - curator.getCtime(executorSystemTimePath));
            int maxTimeDiffSeconds = 10;
            if (timeDiff > maxTimeDiffSeconds * 1000L) {
                Long timeDiffSeconds = Long.valueOf(timeDiff / 1000);
                throw new RuntimeException(timeDiffSeconds.intValue() + maxTimeDiffSeconds + "");
            }
        } finally {
            if (curator.checkExists(executorSystemTimePath)) {
                curator.delete(executorSystemTimePath);
            }
        }
        // 启动时检查Executor是否已启用（ExecutorName为判断的唯一标识）
        if (!curator.checkExists(executorNodePath)) {
            curator.merge(executorNodePath, "");
        } else {
            int count = 0;
            do {
                if (!curator.checkExists(executorIpNodePath)) {
                    return;
                }
                log.info("{}/ip node found. try to sleep and wait for this node disappear。", executorIpNodePath);
                sleep();
            } while (++count <= WAIT_FOR_IP_NODE_DISAPPEAR_COUNT);
            throw new Exception("the executor (" + executorId + ") is running, cannot running the instance twice。");
        }
    }

    private void registerExecutor0() throws Exception {
        // 持久化瞬时ip节点
        curator.ephemeralCreate(executorIpNodePath, HostUtil.cachedIpAddress);
        List<String> taskIdList = new ArrayList<>();
        String taskIds = Joiner.on(Symbols.COMMA).join(taskIdList);
        curator.merge(executorTaskIdListNodePath, taskIds);
        // 持久化最近启动时间节点
        String lastBeginTimeNodePath = executorNodePath + "/lastBeginTime";
        curator.merge(lastBeginTimeNodePath, String.valueOf(System.currentTimeMillis()));
    }

    /**
     * 注销Executor
     */
    public void unRegisterExecutor() {
        unRegisterExecutor0();
    }

    private void unRegisterExecutor0() {
        try {
            if (curator != null && curator.isConnected() && executorIpNodePath != null) {
                log.info("{} is going to delete its ip node {}", executorId, executorIpNodePath);
                curator.delete(executorIpNodePath);
            }
        } catch (Throwable t) {
            log.info(t.toString(), t);
        }
    }
    //##<!-我是优美的功能模块分割线->##

    /**
     * 注册$/Jobs的watcher，响应添加任务的事件，初始化任务。注意，会响应已经存在的任务。
     */
    public void registerTaskWatcher(Executor executor) throws Exception {
        if (taskInitService != null) {
            taskInitService.shutdown();
        }
        taskInitService = new TaskExecutorInitAndTaskItemAssignListenerService(executor, this);
        taskInitService.start();
    }

    /**
     * 销毁监听添加任务的watcher。关闭正在初始化任务的线程，直到其结束。
     */
    public void unRegisterTaskWatcher() {
        if (taskInitService != null) {
            taskInitService.shutdown();
        }
    }


}
